package com.atguigu.chapter11.window;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Over;
import org.apache.flink.table.api.OverWindow;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.*;

/**
 * Author: Pepsi
 * Date: 2023/8/24
 * Desc:
 */
public class Flink06_Window_Over {
    public static void main(String[] args) {

        // 获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<WaterSensor> stream = env.fromElements(
                new WaterSensor("sensor_1", 1000L, 10),
                new WaterSensor("sensor_1", 2000L, 20),
                new WaterSensor("sensor_2", 3000L, 30),
                new WaterSensor("sensor_1", 4000L, 40),
                new WaterSensor("sensor_1", 5000L, 50),
                new WaterSensor("sensor_1", 6000L, 60)
        )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((ws,ts)->ws.getTs())
                );

        // 获取表执行环境，在这之前要获取流的执行环境作为参数传过来
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 把流转换成表
        Table table = tEnv.fromDataStream(stream, $("id"), $("ts"), $("vc"), $("et").rowtime());

        // over(partition by id order by et asc rows between unbounded preceding and current row)
//        OverWindow w = Over.partitionBy($("id")).orderBy($("et")).preceding(Expressions.UNBOUNDED_ROW).following(Expressions.CURRENT_ROW).as("w");

        // over(partition by id order by et asc rows between 1 preceding and current row)   往前算一行
//        OverWindow w = Over.partitionBy($("id")).orderBy($("et")).preceding(rowInterval(1L)).following(CURRENT_ROW).as("w");


        // range 和 row 的区别就是时间一样的话，range会进入同一个窗口
//        OverWindow w = Over.partitionBy($("id")).orderBy($("et")).preceding(UNBOUNDED_RANGE).following(CURRENT_RANGE).as("w");

        // 往前算2s
//        OverWindow w = Over.partitionBy($("id")).orderBy($("et")).preceding(lit(2).second()).following(CURRENT_RANGE).as("w");

        // 不写，等价于 ： over(partition by id order by et asc range between unbounded preceding and current range)
        OverWindow w = Over.partitionBy($("id")).orderBy($("et")).as("w");

        table
                .window(w)
                .select($("id"),$("ts"),$("vc"),$("et"),$("vc").sum().over($("w")).as("sum_vc"))
                .execute()
                .print();



    }
}

/*

sum(vc) over(partition by id order by et asc rows between unbounded preceding and current row)


* */
